package com.bml.architect.base;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 轻量级的拉取消息，不用自己管理offset
 * 但是一般在生产中还是需要自己管理offset的，因为可以自己手动的去重复消费，自己来控制
 * 如果说使用这种模式，就不能重复消费，就是可以自己调整offset的位置就可以实现重复消费了
 */
public class LitePullConsumerSubscribe {
    public static void main(String[] args) {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumerPushGrpoup");
        try {
            consumer.setNamesrvAddr(MqConstants.getNameServerAddr());
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.subscribe("TopicTest", "*");
            consumer.start();
            for (; ; ) {
                List<MessageExt> poll = consumer.poll();
                System.out.printf("%s%n", poll);
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
